package com.taobao.job;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;

import java.io.IOException;
import java.text.DecimalFormat;
import java.util.*;

/**
 * Created by IntelliJ IDEA.
 * User: caoling
 * Date: 11-3-11
 * Time: 11:33
 * To change this template use File | Settings | File Templates.
 */
public class ItemProcess {

    public static final String underLine = "_";
    public static final String colon = ":";
    public static final String space = " ";

    public static void main(String[] args)
            throws Exception {
        JobConf conf = new JobConf(ItemProcess.class);
        conf.setJobName("ItemProcess");
        conf.setOutputKeyClass(Text.class);
        conf.setOutputValueClass(Text.class);

        conf.setMapperClass(Map.class);
        conf.setReducerClass(Reduce.class);

        conf.setMapOutputKeyClass(TextInt.class);
        conf.setMapOutputValueClass(Text.class);

        conf.setInputFormat(TextInputFormat.class);
        conf.setOutputFormat(TextOutputFormat.class);

        conf.setOutputKeyComparatorClass(TextIntComparator.class);
        conf.setOutputValueGroupingComparator(TextComparator.class);
        conf.setPartitionerClass(PartitionByText.class);

        String outputPath = "itemRaw";// + System.currentTimeMillis();
        FileInputFormat.setInputPaths(conf, new Path[]{new Path("data")});
        FileOutputFormat.setOutputPath(conf, new Path(outputPath));

        JobClient.runJob(conf);
    }


    public static class TextIntComparator extends WritableComparator {

        protected TextIntComparator(Class<? extends WritableComparable> keyClass) {
            super(keyClass);
        }

        protected TextIntComparator(Class<? extends WritableComparable> keyClass, boolean createInstances) {
            super(keyClass, createInstances);
        }

        public TextIntComparator() {
            super(TextInt.class, true);
        }


        @Override
        public int compare(WritableComparable a, WritableComparable b) {
            TextInt o1 = (TextInt) a;
            TextInt o2 = (TextInt) b;
            if (!o1.getFirst().equals(o2.getFirst())) {
                return o1.getFirst().compareTo(o2.getFirst());
            } else {
                return o2.getSecond() - o1.getSecond(); //reverse
            }
        }
    }

    public static class TextComparator extends WritableComparator {
        public TextComparator() {
            super(TextInt.class, true);
        }

        @Override
        public int compare(WritableComparable a, WritableComparable b) {
            TextInt o1 = (TextInt) a;
            TextInt o2 = (TextInt) b;
            return o1.getFirst().compareTo(o2.getFirst());
        }
    }

    public static class PartitionByText implements Partitioner<TextInt, IntWritable> {
        @Override
        public int getPartition(TextInt key, IntWritable value, int numPartitions) {
            return (key.getFirst().hashCode() & Integer.MAX_VALUE) % numPartitions;
        }

        @Override
        public void configure(JobConf job) {
        }
    }

    public static class Reduce extends MapReduceBase
            implements Reducer<TextInt, Text, Text, Text> {
        public void reduce(TextInt key, Iterator<Text> values, OutputCollector<Text, Text> output, Reporter reporter)
                throws IOException {
            StringBuffer sb = new StringBuffer();
            int sum = 0;
            while ((values.hasNext()) && (sum++ < 5)) {
                sb.append(values.next().toString()).append("\t");
            }
            output.collect(new Text(key.getFirst()), new Text(sb.toString()));
        }
    }

    public static class Map extends MapReduceBase
            implements Mapper<LongWritable, Text, TextInt, Text> {
        public void map(LongWritable key, Text value, OutputCollector<TextInt, Text> output, Reporter reporter)
                throws IOException {
            String auctionInfo = value.toString();
            List<String> itemInfo = generateItemInfo(auctionInfo);
            String cId = (String) itemInfo.get(1);
            String pvs = (String) itemInfo.get(2);
            double score = Double.valueOf((String) itemInfo.get(3)).doubleValue();
            if ((StringUtils.isNotBlank(cId)) && (StringUtils.isNotBlank(pvs))) {
                StringTokenizer st = new StringTokenizer(pvs);
                while (st.hasMoreTokens()) {
                    String pv = st.nextToken();
                    ItemInfoDO itemDO = new ItemInfoDO(score, (String) itemInfo.get(0));
                    TextInt td = new TextInt(cId + "-" + pv, Double.valueOf(score).intValue());
                    output.collect(td, new Text(itemInfo.get(0)));
                }
            }
        }

        private static String mergePV(String aPV, String spuPV, String spuAPV) {
            Set pvSet = new HashSet();
            addPV(pvSet, aPV);
            addPV(pvSet, spuPV);
            addPV(pvSet, spuAPV);

            return StringUtils.join(pvSet.toArray(), " ");
        }

        private static void addPV(Set<String> pvSet, String pvString) {
            if (StringUtils.isNotEmpty(pvString)) {
                String[] pvs = pvString.split(";");
                for (String pv : pvs)
                    pvSet.add(pv.replace(':', '_'));
            }
        }

        private static List<String> generateItemInfo(String rawItem) {
            List<String> result = new ArrayList<String>();
            String[] infos = rawItem.split("\001");

            StringBuffer itemInfo = new StringBuffer();
            String aId = infos[0];
            itemInfo.append(aId).append(colon);
            String uId = infos[1];
            itemInfo.append(uId).append(colon);
            String score = infos[2];
//            String spuId = infos[3];
//            itemInfo.append(spuId).append(colon);
            String price = infos[4];
            itemInfo.append(price).append(colon);
            String pv = mergePV(infos[10], infos[11], infos[12]);
            itemInfo.append(pv).append(colon);
            String catId = infos[19];
            itemInfo.append((infos[14] + space + infos[15] + space + infos[16] + space + infos[17]).trim()).append(colon);
//            String catPath = (infos[14] + underLine + infos[15] + underLine + infos[16] + underLine + infos[17]).trim();


            String sold30Times = infos[5];
            itemInfo.append(sold30Times).append(colon);
            String sold1Times = infos[6];
            itemInfo.append(sold1Times).append(colon);
            String sold30Num = infos[7];
            itemInfo.append(sold30Num).append(colon);
            String sold1Num = infos[8];
            itemInfo.append(sold1Num).append(colon);
            String collectTime = infos[9];
            itemInfo.append(collectTime).append(colon);
            String uv = infos[13];
            itemInfo.append(uv).append(colon);


            result.add(itemInfo.toString());// + "====" + new DecimalFormat("0.00").format(Double.valueOf(score)) + "====");
            result.add(catId);
            result.add(pv);
            result.add(score);

            return result;
        }
    }
}
